AQS同步机制源码分析(二)

本篇我们主要分析AQS独占模式的源码，关于AQS的独占模式我们上一篇有所介绍。主要这里我们介绍acquire和release部分的代码。这会涉及到AQS的阻塞唤醒机制，还有其维护的FIFO队列。

阻塞过程

独占模式下的阻塞过程

public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
}

tryAcquire返回boolean值，true代表状态更新成功线程继续，否则当前线程需要阻塞，并添加到队列中。这里addWaiter会为当前线程创建Nodj节点并添加到队列中。因为是独占模式节点模式为Node.EXCLUSIVE。

private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    enq(node);
    return node;
}

addWaiter中首先为线程创建node节点,如果tail不为null说明队列不为空，这里先尝试通过CAS将node添加到队尾，然后返回，如果CAS尝试失败，则通过调用enq添加。

private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        if (t == null) { // Must initialize
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

首先enq是一个for循环，这里可以保证node节点一定可以添加到等待队列。首先判断队列是否为空，如果是则new一个Node节点作为当前空队列的头节点，同时将尾节点也指向它，然后在下个循环中同样通过CAS添加当前node节点到队列中。即使失败也通过循环再次尝试添加，直到成功。

我们看看AQS中队列的节点信息

/**
     * Head of the wait queue, lazily initialized.  Except for
     * initialization, it is modified only via method setHead.  Note:
     * If head exists, its waitStatus is guaranteed not to be
     * CANCELLED.
     */
    private transient volatile Node head;

    /**
     * Tail of the wait queue, lazily initialized.  Modified only via
     * method enq to add new wait node.
     */
    private transient volatile Node tail;

AQS队列是通过head和tail节点来维护的，其中Node节点分别有前驱和后继节点。
它是一个简单的队列结构，而保证线程节点能够正确添加到队列中正是基于CAS，这使得它是一个非阻塞式的队列。

final boolean acquireQueued(final Node node, int arg){
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

添加Node到队列中后，有可能会阻塞当前线程，这里获取当前node的前驱，这个前驱节点如果等于head,说明队列中的头节点所代表的线程已经执行完毕release了，当前线程是从parck中解放出来的，这时候当前线程需要通过tryAcquire更新抢占锁，如果抢到了就将当前线程的node节点作为队列的头节点即head，head节点代表的线程是当前正在运行的线程。当然如果p不等于head了，说明它前面还有等待的线程再等待，需要注意的这个队列是一个非阻塞的FIFO队列，所以它需要等待前面的线程执行完毕，因此进入通过shouldParkAfterFailedAcquire判断是否应该park当前线程，返回true表示应该阻塞线程，然后通过parkAndCheckInterrupt来阻塞当前线程。

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        /*
            * This node has already set status asking a release
            * to signal it, so it can safely park.
            */
        return true;
    if (ws > 0) {
        /*
            * Predecessor was cancelled. Skip over predecessors and
            * indicate retry.
            */
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        /*
            * waitStatus must be 0 or PROPAGATE.  Indicate that we
            * need a signal, but don't park yet.  Caller will need to
            * retry to make sure it cannot acquire before parking.
            */
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

这里只有当node的前驱节点的waitStatus为Node.SIGNAL时才会返回true，此时表示前驱节点会unPark node节点的线程，所以可以park这个线程。如果前驱节点不满足这个条件，就需要查找一个不为CANCELED的节点作为node的前驱，并更新它的waitStatus为SIGNAL。

private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        return Thread.interrupted();
}

线程的阻塞是在parkAndCheckInterrupt中进行的，阻塞使用了LockSupport的park。这样当前线程就阻塞在acquireQueued的for循环中等待被唤醒。

共享模式下的阻塞过程

public final void acquireShared(int arg) {
        if (tryAcquireShared(arg) < 0)
            doAcquireShared(arg);
}

在共享模式下的逻辑类似于独占模式，tryAcquireShared返回负值代表未获取到同步状态需要阻塞，这里是通过doAcquireShared来完成的。

private void doAcquireShared(int arg) {
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}
```   

不同于独占模式下的阻塞模式，这里首先针对当前线程创建Node.SHARED节点，使用addWaiter添加到阻塞队列中，随后的逻辑基本和独占模式类似，即通过一个循环来判断当前节点是否满足唤醒的条件:1 当前节点是阻塞队列的第一个节点，且再次通过tryAcqurireShared获取到了同步状态，对于共享模式来说即满足 r>=0，随后将本节点node设置为头结点根据需要设置中断返回。阻塞的逻辑同独占模式下，这里就不再赘述。

## 唤醒操作
```java
public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
}

当子类的实现tryRelease返回true表示释放了同步状态，这时候就可以唤醒当前node节点所代表线程的后继节点了。这一步是通过unparkSuccessor实现的。

private void unparkSuccessor(Node node) {
    /*
        * If status is negative (i.e., possibly needing signal) try
        * to clear in anticipation of signalling.  It is OK if this
        * fails or if status is changed by 
        waiting thread.
        */
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    /*
        * Thread to unpark is held in successor, which is normally
        * just the next node.  But if cancelled or apparently null,
        * traverse backwards from tail to find the actual
        * non-cancelled successor.
        */
    Node s = node.next;
    if (s == null || s.waitStatus > 0) {
        s = null;
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    if (s != null)
        LockSupport.unpark(s.thread);
}

这里取到当前队列的下一个node节点，并通过LockSupport.unpark解除相应线程的阻塞状态。